package com.jsu.listener;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.jsu.config.RabbitMqConfig;
import com.jsu.config.RedisCache;
import com.jsu.mapper.ActivitySubMapper;
import com.jsu.mapper.AgendaSubMapper;
import com.jsu.server.Mail;
import com.jsu.server.WebSocketServer;
import com.rabbitmq.client.Channel;
import com.jsu.client.UserClient;
import org.example.model.activity.pojo.ActivitySub;
import org.example.model.activity.pojo.AgendaSub;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

@Component
public class MessageConsumer {
    @Autowired
    private AgendaSubMapper agendaSubMapper;
    @Autowired
    private ActivitySubMapper activitySubMapper;
    @Autowired
    private RedisCache redisCache;
    @Autowired(required = false)
    private UserClient userClient;



    @RabbitListener(queues = RabbitMqConfig.FANOUT_QUEUE_NAME)
    public void sendAgendaSub(Message message, Channel channel) throws IOException, TimeoutException {
        //拿到议程号
        String[] part = new String(message.getBody()).split(":");
        String agendaName = part[0];
        String agendaId = part[1];
        //拿到订阅该议程的userId
        LambdaQueryWrapper<AgendaSub> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        lambdaQueryWrapper.eq(AgendaSub::getAgendaId,agendaId);
        List<AgendaSub> agendaSubs = agendaSubMapper.selectList(lambdaQueryWrapper);
        String mes = "您订阅的直播："+agendaName+"，已经开始直播啦！";

        //拿到在线用户
        ConcurrentHashMap<String, WebSocketServer> webSocketMap = WebSocketServer.webSocketMap;

        for (AgendaSub agendaSub : agendaSubs) {
            if (webSocketMap.containsKey(agendaSub.getAgendaId().toString())){
                //单点信息发送
                WebSocketServer.sendInfo(mes,agendaSub.getUserId().toString());
            }
            else {
                String email = (String) userClient.getEmail(agendaSub.getUserId()).getData();
                if (StringUtils.hasLength(email)){
                    Mail.sendMail(email,"您订阅的直播已经开始了。","这是来自西湖论剑的直播订阅提醒");
                }

            }
        }
        channel.close();
    }

    @RabbitListener(queues = RabbitMqConfig.FANOUT_QUEUE_NAME)
    public void sendActivitySub(Message message, Channel channel) throws IOException, TimeoutException {
        //拿到议程号
        String[] part = new String(message.getBody()).split(":");
        String activityName = part[0];
        String activityId = part[1];
        //拿到订阅该议程的userId
        LambdaQueryWrapper<ActivitySub> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        lambdaQueryWrapper.eq(ActivitySub::getActivityId,activityId);
        List<ActivitySub> activitySubs = activitySubMapper.selectList(lambdaQueryWrapper);
        String text = "您订阅的直播："+activityName+"，已经开始直播了。";

        //拿到在线用户
        ConcurrentHashMap<String, WebSocketServer> webSocketMap = WebSocketServer.webSocketMap;

        for (ActivitySub activitySub : activitySubs) {
            if (webSocketMap.containsKey(activitySub.getActivityId().toString())){
                //单点信息发送
                WebSocketServer.sendInfo(text,activitySub.getUserId().toString());
            }
            else {
                String email = (String) userClient.getEmail(activitySub.getUserId()).getData();
                if (StringUtils.hasLength(email)){
                    Mail.sendMail(email,text,"这是来自西湖论剑的直播订阅提醒");
                }
            }
        }
        channel.close();
    }

    @RabbitListener(queues = RabbitMqConfig.DELAY_QUEUE_NAME)
    public void sendToAdmin(Message message,Channel channel) throws IOException {
        WebSocketServer.sendToAdminPassage();
        //手动确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

}
